{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "9fb40173",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "链式处理演示\n",
      "------------------------------\n",
      "输入项目信息:\n",
      "\n",
      "    移动应用开发项目，进度60%\n",
      "    完成：UI设计、用户注册登录、核心功能开发\n",
      "    待完成：支付功能、推送通知、性能优化\n",
      "    风险：iOS审核可能延期，安卓兼容性问题\n",
      "    \n",
      "\n",
      "执行链式处理...\n",
      "最终结果:\n",
      "  project_name: 移动应用开发项目\n",
      "  progress: 项目整体进度已完成60%，主要模块开发工作已完成，进入功能完善与优化阶段。\n",
      "  completed_tasks: ['UI设计', '用户注册登录功能开发', '核心功能开发']\n",
      "  pending_tasks: ['支付功能集成与测试', '推送通知功能开发', '应用性能优化']\n",
      "  risks: ['iOS应用审核流程可能延期，存在影响上线时间的风险', '安卓设备兼容性问题需要进一步排查与修复，可能影响用户体验']\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "LangChain自定义模板 - 项目报告场景\n",
    "\"\"\"\n",
    "from typing import Any, Dict, List\n",
    "from pydantic import BaseModel, Field\n",
    "from langchain_core.prompts import StringPromptTemplate\n",
    "from langchain_core.output_parsers import BaseOutputParser\n",
    "from langchain_community.llms import Tongyi\n",
    "\n",
    "# 项目报告数据模型\n",
    "class ProjectReport(BaseModel):\n",
    "    project_name: str = Field(description=\"项目名称\")\n",
    "    progress: str = Field(description=\"进度状态\")\n",
    "    completed_tasks: List[str] = Field(description=\"已完成任务\")\n",
    "    pending_tasks: List[str] = Field(description=\"待完成任务\")\n",
    "    risks: List[str] = Field(description=\"风险点\")\n",
    "\n",
    "# 自定义项目报告模板\n",
    "class ProjectReportTemplate(StringPromptTemplate):\n",
    "    \"\"\"项目报告自定义模板\"\"\"\n",
    "    \n",
    "    language_style: str = Field(description=\"语言风格\", default=\"professional\")\n",
    "    \n",
    "    def __init__(self, **kwargs):\n",
    "        kwargs.setdefault('input_variables', ['input_text'])\n",
    "        super().__init__(**kwargs)\n",
    "    \n",
    "    def format(self, **kwargs: Any) -> str:\n",
    "        \"\"\"格式化项目报告模板\"\"\"\n",
    "        input_text = kwargs.get(\"input_text\", \"\")\n",
    "        \n",
    "        return f\"\"\"作为项目经理，请分析以下项目信息并生成状态报告：\n",
    "\n",
    "                项目信息：\n",
    "                {input_text}\n",
    "\n",
    "                请按以下JSON格式输出项目报告：\n",
    "                {{\n",
    "                \"project_name\": \"项目名称\",\n",
    "                \"progress\": \"进度状态\",\n",
    "                \"completed_tasks\": [\"已完成任务1\", \"已完成任务2\"],\n",
    "                \"pending_tasks\": [\"待完成任务1\", \"待完成任务2\"],\n",
    "                \"risks\": [\"风险点1\", \"风险点2\"]\n",
    "                }}\n",
    "\n",
    "                报告要求：\n",
    "                1. 客观评估项目当前状态\n",
    "                2. 识别关键里程碑和交付物\n",
    "                3. 预警潜在风险和阻碍\n",
    "                4. 采用{self.language_style}的汇报风格\"\"\"\n",
    "\n",
    "# 输出解析器\n",
    "class ProjectParser(BaseOutputParser[Dict[str, Any]]):\n",
    "    \"\"\"项目报告解析器\"\"\"\n",
    "    \n",
    "    def parse(self, text: str) -> Dict[str, Any]:\n",
    "        import json\n",
    "        import re\n",
    "        \n",
    "        # 提取JSON\n",
    "        json_match = re.search(r'\\{.*\\}', text, re.DOTALL)\n",
    "        json_str = json_match.group(0) if json_match else text\n",
    "        \n",
    "        # 解析并验证\n",
    "        data = json.loads(json_str)\n",
    "        validated = ProjectReport(**data)\n",
    "        return validated.model_dump()\n",
    "    \n",
    "    def get_format_instructions(self) -> str:\n",
    "        return \"请以JSON格式输出项目报告\"\n",
    "\n",
    "# 调用通义千问\n",
    "def demo_with_tongyi():\n",
    "    \"\"\"使用通义千问实际调用演示\"\"\"\n",
    "    print(\"项目报告自定义模板 - 通义千问实际调用\")\n",
    "    print(\"-\" * 40)\n",
    "    \n",
    "    # 初始化通义千问\n",
    "    llm = Tongyi(\n",
    "        model_name=\"qwen-max\",\n",
    "        temperature=0.1,\n",
    "        top_p=0.8\n",
    "    )\n",
    "    \n",
    "    # 创建模板和解析器\n",
    "    template = ProjectReportTemplate(language_style=\"executive\")\n",
    "    parser = ProjectParser()\n",
    "    \n",
    "    # 测试项目信息\n",
    "    project_info = \"\"\"\n",
    "    AI客服系统开发项目，启动3个月\n",
    "    已完成：需求分析、技术选型、基础架构搭建、自然语言处理模块\n",
    "    正在进行：对话管理系统、知识库集成\n",
    "    计划完成：用户界面开发、系统测试、部署上线\n",
    "    当前问题：知识库数据质量不够，影响回答准确性\n",
    "    \"\"\"\n",
    "    \n",
    "    # 生成提示\n",
    "    prompt = template.format(input_text=project_info)\n",
    "    print(\"生成的提示:\")\n",
    "    print(prompt[:200] + \"...\")\n",
    "    \n",
    "    # 实际调用通义千问\n",
    "    print(\"\\n正在调用通义千问...\")\n",
    "    response = llm.invoke(prompt)\n",
    "    print(\"\\n通义千问响应:\")\n",
    "    print(response)\n",
    "    \n",
    "    # 解析结果\n",
    "    print(\"\\n解析结果:\")\n",
    "    result = parser.parse(response)\n",
    "    for key, value in result.items():\n",
    "        print(f\"  {key}: {value}\")\n",
    "\n",
    "# 创建完整的处理链\n",
    "def create_project_chain():\n",
    "    \"\"\"创建项目报告处理链\"\"\"\n",
    "    llm = Tongyi(model_name=\"qwen-max\", temperature=0.1)\n",
    "    template = ProjectReportTemplate(language_style=\"professional\")\n",
    "    parser = ProjectParser()\n",
    "    \n",
    "    # 创建链式处理\n",
    "    chain = template | llm | parser\n",
    "    return chain\n",
    "\n",
    "def demo_chain():\n",
    "    \"\"\"演示链式处理\"\"\"\n",
    "    print(\"\\n链式处理演示\")\n",
    "    print(\"-\" * 30)\n",
    "    \n",
    "    chain = create_project_chain()\n",
    "    \n",
    "    project_info = \"\"\"\n",
    "    移动应用开发项目，进度60%\n",
    "    完成：UI设计、用户注册登录、核心功能开发\n",
    "    待完成：支付功能、推送通知、性能优化\n",
    "    风险：iOS审核可能延期，安卓兼容性问题\n",
    "    \"\"\"\n",
    "    \n",
    "    print(\"输入项目信息:\")\n",
    "    print(project_info)\n",
    "    \n",
    "    print(\"\\n执行链式处理...\")\n",
    "    result = chain.invoke({\"input_text\": project_info})\n",
    "    \n",
    "    print(\"最终结果:\")\n",
    "    for key, value in result.items():\n",
    "        print(f\"  {key}: {value}\")\n",
    "\n",
    "def main():\n",
    "    \"\"\"主函数\"\"\"\n",
    "    # demo_with_tongyi()\n",
    "    demo_chain()\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "MLOps",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
